﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Kugar.Core;
using Kugar.Core.BaseStruct;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Framing;
using Kugar.Core.ExtMethod;

namespace Kugar.Tool.RemoteRPC
{
    public class RemoteRPC
    {
        private HashSet<string> _handerServices = new HashSet<string>(StringComparer.CurrentCultureIgnoreCase);
        private IConnection _conn = null;
        private string _serverID = "";
        private IModel _channel = null;
        private IModel _dlxChannel = null;
        private string _currentQueueName = "";
        private EventingBasicConsumer _consumer = null;
        private EventingBasicConsumer _dlxConsumer = null; //用于处理死信队列的消费者
        private Dictionary<string, TimeoutAwaiter<JObject>> _awaiters = new Dictionary<string, TimeoutAwaiter<JObject>>();

        private const string Exchange = "Kugar_RemoteRPC";
        private const string DLXExchnage = "Kugar_RemoteRPC_DLX";
        private const string DLXRoutingKey = "DLX_RPC";

        private string _dlxQueueName = "";  //用于处理死信队列的队列名

        /*
         1. Exchange:  RemoteRPC   RouteKey:  {ServiceName}_RemoteCall      远程调用指定服务
                                              RPC_Queue_{_serverID}         特定服务器的队列

                                   QueueName:  {ServiceName}_Queue   指定服务名称
             */

        public RemoteRPC(IConnection conn, string serverID = null)
        {
            if (string.IsNullOrWhiteSpace(serverID))
            {
                serverID = ShortGuid.NewGuid().ToString();
            }

            _conn = conn;
            _serverID = serverID;

            _currentQueueName = $"RPC_Queue_{_serverID}";
            _dlxQueueName = $"DLX_RPC_{_serverID}";

            //定义DLX死信队列所需的信息
            _dlxChannel = _conn.CreateModel();
            _dlxChannel.ExchangeDeclare(DLXExchnage, ExchangeType.Fanout, false, false, null);
            _dlxChannel.QueueDeclare(_dlxQueueName, false, false, true, null);
            _dlxChannel.QueueBind(_dlxQueueName, DLXExchnage, DLXRoutingKey);

            _dlxConsumer = new EventingBasicConsumer(_dlxChannel);
            _dlxConsumer.Received += dlxConsumer_Received;
            _dlxChannel.BasicConsume(_dlxQueueName, false, _dlxConsumer);

            //定义接收回复信息的队列信息
            _channel = _conn.CreateModel();
            _channel.BasicReturn += channel_BasicReturn;

            _channel.ExchangeDeclare(Exchange, ExchangeType.Direct, true, false, null);

            _channel.QueueDeclare(_currentQueueName, false, false, true, null);

            _channel.QueueBind(_currentQueueName, Exchange, _currentQueueName);
            
            _consumer = new EventingBasicConsumer(_channel);
            
            _consumer.Received += consumer_Received;
            _channel.BasicConsume(_currentQueueName, false, _consumer);
            //_channel.BasicQos(0,2,false);
        }



        public void AddServiceHander(string serviceName)
        {
            if (_handerServices.Add(serviceName))
            {
                var queueName = $"{serviceName}_Queue";

                _channel.QueueDeclare(queueName, false, false, true, new Dictionary<string, object>()
                {
                    ["x-dead-letter-exchange"] = DLXExchnage,
                    ["x-dead-letter-routing-key"] = DLXRoutingKey
                });
                _channel.QueueBind(queueName, Exchange, $"{serviceName}_RemoteCall");
                _channel.BasicConsume(queueName, false, _consumer);

                //_dlxChannel.QueueBind(_currentQueueName,"Kugar_RemoteRPC_DLX",$"{serviceName}_RemoteCall");
            }
        }

        public async Task<JObject> CallAsync(string serviceName, JObject arguments)
        {
            var callID = Guid.NewGuid().ToString();

            var arg = new JObject()
            {
                ["Arguments"] = arguments,
                ["ServiceName"] = serviceName,
                ["RelyID"] = callID,
                ["RelyQueueName"] = _currentQueueName
            };

            var a = new TimeoutAwaiter<JObject>(30000);
            _awaiters.Add(callID, a);

            var routeKey = $"{serviceName}_RemoteCall";

            var prop = new BasicProperties();
            prop.ReplyTo = _currentQueueName;
            prop.CorrelationId = callID;
            prop.Expiration = "20000";

            _channel.BasicPublish(Exchange,
                routeKey,
                true,
                prop,
                Encoding.UTF8.GetBytes(arg.ToString(Formatting.None))
                );

            try
            {
                var value = await a;

                var error = a.GetError();

                if (error != null)
                {
                    throw error;
                }
                else
                {
                    return value;
                }
            }
            catch (Exception)
            {

                throw;
            }
            finally
            {
                _awaiters.Remove(callID);
            }

        }

        public async Task<T> CallAsync<T>(string serviceName, JObject arguments)
        {
            return (await CallAsync(serviceName, arguments)).ToObject<T>();
        }

        public JObject Call(string serviceName, JObject arguments)
        {
            return this.CallAsync(serviceName, arguments).Result;
        }

        public T Call<T>(string serviceName, JObject arguments)
        {
            var v = Call(serviceName, arguments);

            return v.ToObject<T>();
        }

        public event EventHandler<ServiceInvokeEventArgs> ServiceInvoke;

        public event EventHandler<ServiceInvokeErrorEventArgs> ServiceCallError;

        /// <summary>
        /// 当接收到一条来自消息系统的远程调用时,出发该函数,进行处理
        /// </summary>
        /// <param name="e"></param>
        private void onRPCCallTrigger(BasicDeliverEventArgs e)
        {
            var index = e.RoutingKey.LastIndexOf('_');

            var eventName = e.RoutingKey.Substring(0, index);

            try
            {
                if (_handerServices.Contains(eventName))
                {
                    if (ServiceInvoke != null)
                    {
                        onServiceInvoke(e);

                        //Task.Factory.StartNew(onServiceInvoke, e);
                    }                    
                }
            }
            catch (Exception ex)
            {
                //_channel.BasicRecover(true);

                //_channel.BasicReject(e.DeliveryTag, true);

                //throw;
            }
            finally
            {
                _channel.BasicAck(e.DeliveryTag, false);
            }

        }

        private void onServiceInvoke(object state)
        {
            var e = (BasicDeliverEventArgs)state;

            Exception error = null;
 
            JObject returnValue = null;
            JObject callArgument = null;
            string serviceName = string.Empty;
            try
            {
                var package = JsonConvert.DeserializeObject<ServiceCallPackage>(Encoding.UTF8.GetString(e.Body));
                
                ServiceInvokeEventArgs ea = null;

                callArgument = package.Arguments;
                serviceName = package.ServiceName;
                ea = new ServiceInvokeEventArgs()
                {
                    Arguments = callArgument,
                    ServiceName = serviceName
                };

                ServiceInvoke?.Invoke(this, ea);

                returnValue = ea.ReturnValue;
            }
            catch (Exception ex)
            {
                error = ex;

                if (ServiceCallError!=null)
                {
                    var ea=new ServiceInvokeErrorEventArgs()
                    {
                        Arguments = callArgument,
                        Error = ex,
                        IsReject = false,
                        ServiceName = serviceName
                    };

                    ServiceCallError(this, ea);
                }
            }
            finally
            {
                
            }

            //var args=new ServiceRelyPackage()
            //{
            //    CallerID = package.RelyID,
            //    ReturnValue = ea.ReturnValue,
            //    Error = error
            //};

            var args = new JObject()
            {
                ["CallerID"] = e.BasicProperties.CorrelationId,
                ["ReturnValue"] =returnValue,
                ["Error"] = error == null ? null : JObject.FromObject(error)
            };

            _channel.BasicPublish(Exchange, e.BasicProperties.ReplyTo, new BasicProperties()
            {
                Expiration = "15000",
                CorrelationId = e.BasicProperties.CorrelationId
            }, Encoding.UTF8.GetBytes(args.ToString(Formatting.None)));
        }

        /// <summary>
        /// 当接收到远程数据处理后,回复数据时的处理函数
        /// </summary>
        /// <param name="e"></param>
        private void onRPCRelyReceive(BasicDeliverEventArgs e)
        {
            _channel.BasicAck(e.DeliveryTag, false);

            TimeoutAwaiter<JObject> value;

            if (_awaiters.TryGetValue(e.BasicProperties.CorrelationId, out value))
            {
                var json = JsonConvert.DeserializeObject<ServiceRelyPackage>(Encoding.UTF8.GetString(e.Body));

                _awaiters.Remove(json.CallerID);

                var error = json.Error;

                if (json.Error != null)
                {
                    value.SetError(error);
                }
                else
                {
                    value.SetResult(json.ReturnValue, true);
                }
            }
        }

        private void consumer_Received(object sender, BasicDeliverEventArgs e)
        {
            if (e.RoutingKey.EndsWith("_RemoteCall"))
            {
                onRPCCallTrigger(e);
            }
            else if (e.RoutingKey.StartsWith("RPC_Queue_"))
            {
                onRPCRelyReceive(e);
            }
        }

        /// <summary>
        /// 发生BasicReturn消息时候,如不存在接收队列等
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void channel_BasicReturn(object sender, BasicReturnEventArgs e)
        {
            handlerTimeoutOrBasicReturn(e.BasicProperties?.CorrelationId, e.RoutingKey);
        }

        /// <summary>
        /// 处理死信队列消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void dlxConsumer_Received(object sender, BasicDeliverEventArgs e)
        {
            if (e.BasicProperties.ReplyTo == _currentQueueName)
            {
                handlerTimeoutOrBasicReturn(e.BasicProperties?.CorrelationId, e.RoutingKey);
            }

            _dlxChannel.BasicAck(e.DeliveryTag, false);
        }

        /// <summary>
        /// 发生退信,或者死信队列接受到消息时的处理函数
        /// </summary>
        /// <param name="callID"></param>
        /// <param name="routerKey"></param>
        private void handlerTimeoutOrBasicReturn(string callID, string routerKey)
        {
            TimeoutAwaiter<JObject> tmp;

            if (_awaiters.TryGetValue(callID, out tmp))
            {
                Exception error = null;

                if (routerKey.EndsWith("_RemoteCall")) //发起调用遭退信时,抛出错误
                {
                    error = new NotImplementedException("不存在指定的服务,或服务端未启动");
                }
                else if (routerKey == DLXRoutingKey)
                {
                    error = new TimeoutException("退信:调用超时");
                }


                tmp.SetError(error ?? new Exception("未知错误"));
            }
        }

        private class ServiceCallPackage
        {
            public string ServiceName { set; get; }

            public JObject Arguments { set; get; }

            public string RelyID { set; get; }

            public string RelyQueueName { set; get; }
        }

        public class ServiceRelyPackage
        {
            public JObject ReturnValue { set; get; }

            public string CallerID { set; get; }

            public Exception Error { set; get; }
        }
    }

    public class ServiceInvokeEventArgs : EventArgs
    {
        public string ServiceName { set; get; }

        public JObject Arguments { set; get; }

        public JObject ReturnValue { set; get; }
    }

    public class ServiceInvokeErrorEventArgs : EventArgs
    {
        public string ServiceName { set; get; }

        public bool IsReject { set; get; } = false;

        public JObject Arguments { set; get; }

        public Exception Error { set; get; }

        public JObject ReturnValue { set; get; }
    }
}
